package test.mq;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.junit.FixMethodOrder;
import org.junit.Test;
import hyl.base.mq.active.ActConfig;
import hyl.base.mq.active.IReceive;
import hyl.base.mq.active.MyActiveMq;
import hyl.core.MyDate;
import hyl.core.MyFun;
import hyl.core.run.MyRun;

@FixMethodOrder(org.junit.runners.MethodSorters.NAME_ASCENDING)
public class testactivemq {
	static ActConfig ac = ActConfig.loadFile("mq/activemq.conf");

	// 方案1
	public static void main(String[] arg) throws Exception {
		System.out.println(MyActiveMq.checkConnectConfig(ac));
	}

	// @Test // 方案1
	public void test0() throws Exception {
		MyActiveMq mmp = MyActiveMq.getInstance();
		testsendqueue(mmp);
		MyActiveMq mms = MyActiveMq.getInstance();
		testsub(mms);
		testsub(mms);
		testsub(mms);
		MyDate.sleep(1000);
		// mmp.close();
		// mms.close();
	}

	/*
	 * 这是一个错误,无法实现的例子 订阅写在前会阻塞 订阅写在后又无法订阅到课题 所以测试TOPIC_MODEL 模式 必须采用线程 而且必须 先订阅,后发送
	 */
	// @Test
	public void test1_1() throws Exception {
		MyActiveMq mmp = MyActiveMq.getInstance(ac, "queue/test1", MyActiveMq.TOPIC_MODEL, null);
		MyActiveMq mms = MyActiveMq.getInstance(ac, "queue/test1", MyActiveMq.TOPIC_MODEL, null);
		// testsub(mms);// 此处会阻塞,导致无法执行后续
		testsendqueue(mmp);
		// testsub(mms);// 此处会无法订阅到queue/test1
		MyDate.sleep(9000);
		mmp.close();
	}

	/** 测试应答函数的作用 */
	// @Test
	public void testqueue() throws JMSException {
		MyActiveMq mmp = MyActiveMq.getInstance(ac, "queue/test11", MyActiveMq.QUEUE_MODEL, Session.CLIENT_ACKNOWLEDGE);
		MyActiveMq mms = MyActiveMq.getInstance(ac, "queue/test11", MyActiveMq.QUEUE_MODEL, Session.CLIENT_ACKNOWLEDGE);
		MyRun mr = MyRun.getInstance("测试", 3);
		mr.execute(new Runnable() {
			@Override
			public void run() {
				try {
					System.out.println("收");
					while (true) {
						mms.receive(new IReceive() {
							@Override
							public void run(Object obj, Message msg) {
								System.out.println((String) obj);
								mms.acknowledge(msg);
							}
						});
					}
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		});
		mr.execute(new Runnable() {
			@Override
			public void run() {
				try {
					testsend(mmp);

				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
		MyDate.sleep(9000);
		// mr.close();
	}

	// 测试事务的作用
	// @Test
	public void test4() throws JMSException {
		MyActiveMq mmp = MyActiveMq.getInstance(ac, "topic/test1", MyActiveMq.TOPIC_MODEL, Session.AUTO_ACKNOWLEDGE);
		MyActiveMq mms = MyActiveMq.getInstance(ac, "topic/test1", MyActiveMq.TOPIC_MODEL, Session.SESSION_TRANSACTED);
		MyActiveMq mms2 = MyActiveMq.getInstance(ac, "topic/test1", MyActiveMq.TOPIC_MODEL, Session.SESSION_TRANSACTED);
		MyRun mr = MyRun.getInstance("测试", 4);
		mr.execute(new Runnable() {
			@Override
			public void run() {
				try {
					testsend_commit(mmp);
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		});
		mr.execute(new Runnable() {
			@Override
			public void run() {
				testsub_commit(mms);
			}
		});
		mr.execute(new Runnable() {
			@Override
			public void run() {
				testsub_commit(mms2);
			}
		});
		MyDate.sleep(19000);
		mr.close();
	}

	@Test
	public void test100() throws JMSException {

		MyRun mr = MyRun.getInstance("测试", 3);
		mr.execute(new Runnable() {
			@Override
			public void run() {
				test1receive();
			}
		});
		mr.execute(new Runnable() {
			@Override
			public void run() {
				test2send();
			}
		});
		MyDate.sleep(5000);
	}

	// 、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、、
	public void testsendqueue(MyActiveMq mmq) throws JMSException {
		// MyDate.sleep(1000);
		System.out.println("发送");
		mmq.sendText("aaa");
		mmq.sendText("bbb");
		mmq.sendText("ccc");
		System.out.println("发送..");
	}

	public void testsend(MyActiveMq mmq) throws JMSException {
		System.out.println("发送");
		MyDate.sleep(1000);
		mmq.sendText("aaa");
		MyDate.sleep(3000);
		mmq.sendText("bbb");
		MyDate.sleep(3000);
		mmq.sendText("ccc");
		System.out.println("..完成");
	}

	public void testsend_commit(MyActiveMq mmq) throws JMSException {
		MyDate.sleep(1000);
		mmq.sendText("aaa");
		MyDate.sleep(3000);
		mmq.sendText("bbb");
		MyDate.sleep(3000);
		mmq.sendText("ccc");
		mmq.commit();
		mmq.close();
		System.out.println("发送..完成");
	}

	public void testsendtopic_commit(MyActiveMq mmq) throws JMSException {
		MyDate.sleep(1000);
		mmq.sendText("aaa");
		mmq.sendText("bbb");
		mmq.sendText("ccc");
		mmq.commit();
		mmq.close();
	}

	public void testsendtopic_commit2(MyActiveMq mmq) throws JMSException {
		mmq.sendText("aaa");
		mmq.commit();
		MyDate.sleep(3000);
		mmq.sendText("bbb");
		mmq.commit();
		MyDate.sleep(3000);
		mmq.sendText("ccc");
		mmq.commit();
		mmq.close();
	}

	public void testsub(MyActiveMq mmq) throws Exception {
		String t = mmq.receiveText();
		System.out.println("接收.." + t);
	}

	public void testsub_commit(MyActiveMq mmq) {
		String t;
		try {
			while (true) {
				t = mmq.receiveText();
				System.out.println(t + Thread.currentThread().getName());
			}
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void test1receive() {
		System.out.println("start receive");
	
		MyActiveMq mms;
		try {
			mms = MyActiveMq.getInstance(ac, "TOPIC/test1", MyActiveMq.QUEUE_MODEL, Session.CLIENT_ACKNOWLEDGE);

			Long i = MyFun.getMs();
			boolean flag = true;
			while (flag) {
				String t = mms.receiveText();
				if (t.equals("end"))
					flag = false;
				System.out.println(t);
			}
			long l = MyFun.getMs() - i;
			System.out.println("用时：" + l / 1000);
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	public void test2send() {
		MyDate.sleep(1000);
		System.out.println("start send");
		MyActiveMq mmp;
		try {
			mmp = MyActiveMq.getInstance(ac, "TOPIC/test1", MyActiveMq.QUEUE_MODEL, Session.CLIENT_ACKNOWLEDGE);//null Session.CLIENT_ACKNOWLEDGE

			for (int i = 0; i < 100; i++) {
				mmp.sendText("est" + i);
			}
		//	mmp.sendText("end");
			// mmp.close();
			// mms.close();
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
